package com.hjd.shop.mq;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.mapper.Condition;
import com.hjd.shop.api.service.GoodsNumberLogService;
import com.hjd.shop.api.service.MqConsumerLogService;
import com.hjd.shop.api.service.SGoodsService;
import com.hjd.shop.constant.MqProperties;
import com.hjd.shop.constant.MqState;
import com.hjd.shop.constant.SystemUser;
import com.hjd.shop.entity.GoodsNumberLog;
import com.hjd.shop.entity.MqConsumerLog;
import com.hjd.shop.entity.SGoods;
import com.hjd.shop.pojo.MqEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @Author: hjd
 * @Date: 2020/10/28 0028 22:00
 * @Des: rockmq监听类
 * 1.首先需要实现RocketMQListener接口 并重写 onMessage方法
 *   PS:RocketMQListener<T> 后面的泛型就是消息的生产者发送过来的消息的类型。
 * 2.其中rocketmq的messageModel（消息模型）需要设置为 广播(MessageModel.BROADCASTING)，
 * 因为当 确认订单 发生异常时，其他服务都需要接收到 mq的消息。
 */
@Component
@RocketMQMessageListener(topic = "${mq.order.topic}",consumerGroup = "${mq.order.consumer.group.name}",
        messageModel = MessageModel.BROADCASTING)
@Slf4j
public class OrderCancelMqListener implements RocketMQListener<MessageExt> {
    @Autowired
    private MqConsumerLogService mqConsumerLogService;
    @Autowired
    private SGoodsService goodsService;
    @Autowired
    private GoodsNumberLogService goodsNumberLogService;
    @Value("${mq.order.consumer.group.name}")
    private String groupName;
    @Override
    public void onMessage(MessageExt message) {
         //1.解析mq的消息
        String msgId = message.getMsgId();
        String keys = message.getKeys();
        String body = message.getBody().toString();
        String tags = message.getTags();
        //2.查询rocketmq消息消费的日志
        Condition condition = Condition.create();
        condition.eq("msg_id",msgId)
                .eq("msg_key",keys)
                .eq("msg_tag",tags)
                .eq("group_name",groupName);
        MqConsumerLog result= mqConsumerLogService.selectOne(condition);
        //3.判断消息是否已经被消费过了
        try {
            if (result!=null) {
                //说明消息已经被消费过了
                //此时查看消息的状态
                int consumerStatus = result.getConsumerStatus();
                if (consumerStatus==MqState.CONSUMING.getCode()||consumerStatus==MqState.CONSUMING.getCode()) {
                    //已消费和消费中都是要直接结束
                    log.info("消息："+msgId+"已经处理过了/正在处理中...！");
                    return;
                }
                //消息的处理次数
                int consumerTimes = result.getConsumerTimes();
                if (consumerStatus==MqState.CONSUME_FAIL.getCode()) {
                    //处理失败,获取消息的处理次数，如果大于3不做处理，否则再次处理
                    if(consumerTimes >= MqProperties.CONSUME_MAX_TIMES){
                        log.info("消息："+msgId+"处理次数超过3次，取消处理！");
                        return;
                    }
                    //再次处理消息，使用数据库的乐观锁，改变消息的状态.此处我的做法是
                    //在 消费状态 属性上加上@Version注解
                    MqConsumerLog mqConsumerLog = new MqConsumerLog();
                    mqConsumerLog.setConsumerStatus(MqState.CONSUMING.getCode())
                            .setModifyTime(new Date())
                            .setModifyUser(SystemUser.SYSTEM_USER);
                    boolean updateForBack = mqConsumerLogService.updateById(mqConsumerLog);
                    if (updateForBack) {
                        //未修改成功，其他线程并发修改
                        log.info("请稍后重试！");
                    }
                }
            }else {
                MqConsumerLog newMqConsumerLog = new MqConsumerLog();
                newMqConsumerLog.setMsgTag(tags)
                        .setMsgKey(keys)
                        .setMsgBody(body)
                        .setConsumerStatus(MqState.CONSUMING.getCode())
                        .setMsgId(msgId)
                        .setGroupName(groupName)
                        .setCreateUser(SystemUser.SYSTEM_USER)
                        .setCreateTime(new Date());
                mqConsumerLogService.insert(newMqConsumerLog);
                //4.如果没有被消费过回退库存
                MqEntity mqEntity = JSON.parseObject(body, MqEntity.class);
                SGoods sGoods = goodsService.selectById(mqEntity.getGoodsId());
                sGoods.setGoodsName(sGoods.getGoodsName()+mqEntity.getGoodsNum());
                goodsService.updateById(sGoods);
                //5.记录消费日志
                GoodsNumberLog goodsNumberLog = new GoodsNumberLog();
                goodsNumberLog.setOrderId(mqEntity.getOrderId())
                        .setGoodsId(mqEntity.getGoodsId())
                        .setGoodsNumber(mqEntity.getGoodsNum())
                        .setCreateTime(new Date())
                        .setCreateUser(SystemUser.SYSTEM_USER);
                goodsNumberLogService.insert(goodsNumberLog);
                //6.将消息处理状态改为成功
                newMqConsumerLog.setConsumerStatus(MqState.CONSUMED.getCode());
                mqConsumerLogService.updateById(newMqConsumerLog);
                log.info("订单："+mqEntity.getOrderId()+"回退库存成功!");
            }
        }catch (Exception e){
            log.info("消息:"+msgId+"处理失败!");
            e.printStackTrace();
            //消费失败 如果数据库中有数据那么更新它的失败次数，如果没有则插入
            // 需要向数据库中插入一条该消息的记录，并且消费状态为 消费失败；消费失败次数 为 1
            MqConsumerLog mc=new MqConsumerLog();
            mc.setMsgId(msgId)
                    .setMsgTag(tags)
                    .setMsgKey(keys)
                    .setMsgBody(body)
                    .setConsumerTimes(mc.getConsumerTimes()+1)
                    .setConsumerStatus(MqState.CONSUME_FAIL.getCode())
                    .setCreateUser(SystemUser.SYSTEM_USER)
                    .setCreateTime(new Date());
            MqConsumerLog mqResult = mqConsumerLogService.selectById(msgId);
            if (mqResult==null) {
                mqConsumerLogService.insert(mc);
            }else {
                mqConsumerLogService.updateById(mc);
            }
        }

    }
}
